package com.flink.wc.demo.demo_20241225;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

public class FlinkFromKafkaWirteMysql {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.151.168:9092,192.168.151.169:9092,192.168.151.170:9092");
        properties.setProperty("group.id", "test");
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                //topic
                "test_tan_emn",
                new SimpleStringSchema(),
                properties
        );
        env.setParallelism(1);
        env.addSource(kafkaConsumer).addSink(JdbcSink.sink(
                "insert into flink_test(ip, time, type, api) values(?, ?, ?, ?)",
                // 解析数据到sql中
                new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        System.out.println("receive ====> " + s);
                        // 解析数据
                        String[] elements = s.split(",");
                        // 新增数据
                        preparedStatement.setString(1, elements[0]);
                        preparedStatement.setString(2, elements[1]);
                        preparedStatement.setString(3, elements[2]);
                        preparedStatement.setString(4, elements[3]);
                    }
                },
                // 配置数据库连接选项
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        // MySQL 数据库 URL
                        .withUrl("jdbc:mysql://localhost:3306/mysql")
                        // MySQL JDBC 驱动
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        // 用户名
                        .withUsername("root")
                        // 密码
                        .withPassword("123456")
                        .build()
        ));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
